package com.tang.consumer.netty;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.tang.consumer.connection.ConnectManage;
import com.tang.consumer.domain.Request;
import com.tang.consumer.domain.Response;
import com.tang.consumer.netty.codec.json.JSONDecoder;
import com.tang.consumer.netty.codec.json.JSONEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.net.SocketAddress;
import java.util.concurrent.SynchronousQueue;

/**
 * @author Tang
 * @classname NettyClient
 * @description [  ]
 * @date 2021/1/24 18:02
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyClient {

    @Resource
    public ConnectManage connectManage;

    private EventLoopGroup workGroup = new NioEventLoopGroup(10);
    private Bootstrap bootstrap = new Bootstrap();
    private NettyClientHandler clientHandler;

    public NettyClient() {

        clientHandler = new NettyClientHandler(this.connectManage);
        // 初始化 client
        bootstrap.channel(NioSocketChannel.class)
                .group(workGroup)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new IdleStateHandler(30,30,60));
                        pipeline.addLast(new JSONDecoder());
                        pipeline.addLast(new JSONEncoder());
                        pipeline.addLast(clientHandler);
                    }
                });
    }

    /**
     * 发送 请求
     * @param request
     * @return
     */
    public Object send(Request request) {
        Channel channel = connectManage.chooseChannel();
        if (channel != null && channel.isOpen() && channel.isActive()) {
            // 发送请求
            SynchronousQueue<Object> queue = clientHandler.sendRequest(request,channel);
            try {
                // 阻塞等待结果
                return queue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        Response response = new Response();
        response.setCode(1);
        response.setErrorMsg("未正确连接到服务器,请检查相关配置信息!");
        return JSON.toJSONString(response);
    }

    @PreDestroy
    public void destroy() {
        log.info("Rpc客户端推出...释放资源!");
        workGroup.shutdownGracefully();
    }

    /**
     * 连接 node
     * @param node
     * @return
     * @throws InterruptedException
     */
    public Channel doConnect(SocketAddress node) throws InterruptedException {
        ChannelFuture future = bootstrap.connect(node).sync();
        return future.channel();
    }

}